<?php

declare(strict_types=1);

namespace App\Process;

use App\Service\DeviceMqttService;
use App\Utils\Logger;
use App\Service\MqttClientService;
use Hyperf\Process\AbstractProcess;
use Hyperf\Process\Annotation\Process;

use Hyperf\Utils\Codec\Json;

/**
 * @Process(name="MqttSubscribe")
 */
#[Process(name: 'MqttSubscribe')]
class MqttSubscribe extends AbstractProcess
{
    public $nums = 1;

    public function handle(): void
    {
        $service = $this->container->get(MqttClientService::class);
        $client = $service->client;
        $logger = $this->container->get(Logger::class);
        $service = $this->container->get(DeviceMqttService::class);

        $topics = [
            '/v2/EM/+/rx' => 0
        ];
        $client->subscribe($topics);

        while (true) {
            $buffer = $client->recv();
            if ($buffer === true) {
                $client->subscribe($topics);
                continue;
            }
            $logger->get('mqtt')->notice('recv', [$buffer]);
            try {
                go(function() use ($buffer, $logger, $service, $client) {
                    if (isset($buffer['message']) && is_string($buffer['message'])) {
                        $return = $service->subscribe($buffer['message'], str_replace("rx", "tx", $buffer['topic']));
                        if ($return) {
                            $client->publish(str_replace("rx", "tx", $buffer['topic']), $return);
                        }
                    }
                });
            } catch (\Exception $e) {
                $logger->get()->error($e);
            }
        }
    }
}
